package hydrap2p.proxy.manager;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.util.Map;
import java.util.TreeMap;

public class StreamMultiplexer {
	private OutputStream stream;
	private Map<Integer, byte[]> plexes;
	private Map<Integer, Integer> plexoffsets;
	private final static int CHUNKING=2000;
	private Runner runner;
	public StreamMultiplexer(OutputStream stream)
	{
		this.stream=stream;
		plexes=new TreeMap<Integer, byte[]>();
		plexoffsets=new TreeMap<Integer, Integer>();
		runner=new Runner();
		Thread temp=new Thread(runner, "StreamMultiplexer on "+stream);
		temp.setDaemon(true);
		temp.start();
	}
	
	public void send(byte[] tosend)
	{
		// put the data in the map somewhere
		boolean putted=false;
		int key=0;
		while (!putted)
		{
			key=1;
			synchronized(plexes)
			{
				int tries=255;
				while (plexes.containsKey(key) && tries>0)
				{
					key=(key+1) % 256;
					tries--;
				}
				if (!plexes.containsKey(key))
				{
					plexes.put(key, tosend);
					plexoffsets.put(key, 0);
					putted=true;
					//System.out.println("Posted data of size "+tosend.length+" to send as streamid "+key);
				}
			}
			if (!putted)
			{
				try {
					Thread.sleep(200);
				} catch (InterruptedException e) {
					// Why?
				}
			}
		}
		// wait for the data to be sent
		boolean contains=true;
		while (contains)
		{
			synchronized(plexes)
			{
				if (plexes.get(key)!=tosend)
					contains=false;
				if (contains)
				{
					plexes.notifyAll();
				}
			}
			if (contains)
			{
				try {
					synchronized(tosend)
					{
						tosend.wait(200);
					}
				} catch (InterruptedException e) {
					// Yay?
				}
			}
		}

		//System.out.println("Done sending data of size "+tosend.length+" to send as streamid "+key);
	}
	
	private class Runner implements Runnable
	{
		private boolean running;
		public Runner()
		{
			running=true;
		}
		
		@Override
		public void run() {
			boolean sentflag=false;
			while (running)
			{
				sentflag=false;
				for (int i=0; i<256; i++)
				{
					byte[] buffer;
					int offset;
					int sent;
					synchronized(plexes)
					{
						buffer=plexes.get(i);
						if (buffer==null)
							continue;
						offset=plexoffsets.get(i);
					}
					try {
						sent=Math.min(CHUNKING, buffer.length-offset);
						stream.write(i);
						stream.write(NetworkWriter.serializeInt(sent));
						stream.write(buffer, offset, sent);
						stream.flush();
						sentflag=true;
						offset+=sent;
						synchronized(plexes)
						{
							if (offset==buffer.length)
							{
								plexes.remove(i);
								plexoffsets.remove(i);
								//System.out.println("Sent data of size "+buffer.length+" to send as streamid "+i);
								synchronized(buffer)
								{
									buffer.notifyAll();
								}
							}
							else
							{
								plexoffsets.put(i, offset);
								//System.out.println("Sent "+sent+" data of size "+buffer.length+" to send as streamid "+i);
							}
						}
					} catch (IOException e) {
						shutdown();
						break;
					}
					
				}

				if (!sentflag)
				{
					synchronized(plexes)
					{
						try {
							plexes.wait();
						} catch (InterruptedException e) {
							// 
						}
					}
				}
			}
		}
		
		public void shutdown()
		{
			running=false;
			try {
				stream.close();
			} catch (IOException e1) {
				// Give up
			}
			synchronized(plexes)
			{
				for (int i=0; i<256; i++)
				{
					byte[] temp=plexes.get(i);
					if (temp!=null)
					{
						synchronized(temp)
						{
							temp.notify();
						}
					}
				}
				plexes.notifyAll();
			}
			plexes.clear();
			plexoffsets.clear();

		}
	}
}
